import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.alibaba.fastjson.JSON;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class VehicleDataKafkaProducer {

    // Kafka 配置
    private static final String KAFKA_BOOTSTRAP_SERVERS = "node101:9092";
    private static final String TOPIC_NAME = "vehicle_realtime_data";

    // 车辆状态枚举（模拟互动类型）
    enum VehicleEvent {
        DRIVING, STOPPED, CHARGING, MAINTENANCE
    }

    // 整合五表逻辑的车辆数据模型
    static class VehicleData {
        // 订单表 -> 车辆订单ID
        private String orderId = "ORD" + System.currentTimeMillis();
        // 直播间表 -> 车辆所属车队ID（模拟 room_id）
        private String fleetId = "FLEET" + new Random().nextInt(1000);
        // 互动表 -> 车辆事件类型（模拟 interact_type）
        private VehicleEvent eventType = VehicleEvent.values()[new Random().nextInt(VehicleEvent.values().length)];
        // 流量表 -> 进入区域时间
        private String enterTime = System.currentTimeMillis() + "";
        // 流量表 -> 离开区域时间（可为空）
        private String leaveTime = new Random().nextBoolean() ? (System.currentTimeMillis() + 3600000 + "") : null;
        // 商品表 -> 车辆型号（模拟 product_sku）
        private String vehicleModel = "MODEL-" + UUID.randomUUID().toString().substring(0, 6);
        // 自定义车辆属性：位置
        private String location = getRandomLocation();
        // 自定义车辆属性：速度
        private double speed = new Random().nextDouble() * 120;
        // 原始日志（模拟 raw_*_log）
        private String rawLog = "{\"source\":\"vehicle_sensor\",\"version\":\"1.0\"}";
        // 时间戳（模拟 create_time）
        private String createTime = System.currentTimeMillis() + "";

        // 模拟 GPS 位置
        private String getRandomLocation() {
            return String.format("%.6f,%.6f",
                    30 + new Random().nextDouble() * 2,
                    120 + new Random().nextDouble() * 2);
        }

        // Getter 用于 JSON 序列化
        public String getOrderId() { return orderId; }
        public String getFleetId() { return fleetId; }
        public VehicleEvent getEventType() { return eventType; }
        public String getEnterTime() { return enterTime; }
        public String getLeaveTime() { return leaveTime; }
        public String getVehicleModel() { return vehicleModel; }
        public String getLocation() { return location; }
        public double getSpeed() { return speed; }
        public String getRawLog() { return rawLog; }
        public String getCreateTime() { return createTime; }
    }

    public static void main(String[] args) {
        // Kafka 生产者配置
        Properties props = new Properties();
        props.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            Random random = new Random();
            while (true) {
                // 生成车辆数据
                VehicleData data = new VehicleData();

                // 转换为 JSON 字符串
                String json = JSON.toJSONString(data);

                // 发送到 Kafka
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, json);
                producer.send(record);

                // 打印日志
                System.out.println("发送车辆数据: " + json);

                // 控制发送频率（每秒 1 条）
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }
}